package com.infore.gateway.main.filter;

import cn.hutool.json.JSONUtil;
import com.google.common.net.HttpHeaders;
import com.infore.gateway.main.config.SystemProperties;
import com.infore.gateway.main.constant.GatewayConstant;
import com.infore.gateway.main.model.TOthGatewayRequestLog;
import io.netty.buffer.ByteBufAllocator;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.core.Ordered;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.core.io.buffer.NettyDataBufferFactory;
import org.springframework.http.HttpStatus;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.http.server.reactive.ServerHttpRequestDecorator;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.nio.CharBuffer;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;


@Slf4j
@Component
public class BaseGlobalFilter implements GlobalFilter, Ordered {

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    @Autowired
    private SystemProperties properties;
    // 用于缓存规则
    private final Map<String, Integer> rules = new ConcurrentHashMap<>();
    // 用于缓存计数器
    private final Map<String, AtomicInteger> counters = new ConcurrentHashMap<>();
    //用于缓存访问时间
    private final Map<String, Long> timesmillis = new ConcurrentHashMap<>();

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {

        ServerHttpRequest request = exchange.getRequest();
        ServerHttpResponse response = exchange.getResponse();

        String ip = request.getHeaders().getFirst(HttpHeaders.X_FORWARDED_FOR);
        if (ip == null) {
            ip = request.getRemoteAddress().getAddress().getHostAddress();
        }
        String account = "";
        String methodName = request.getMethodValue();
        String url = request.getURI().getPath();
        String requestBody = null;
        //判断是否为POST请求
        if("POST".equalsIgnoreCase(methodName)) {
            //多次读取request body会报错Only one connection receive subscriber allowed.
            requestBody = resolveBodyFromRequest(request);
            //转发出去，不然会报错Only one connection receive subscriber allowed.
            DataBuffer bodyDataBuffer = stringBuffer(requestBody);
            Flux<DataBuffer> bodyFlux = Flux.just(bodyDataBuffer);
            request = new ServerHttpRequestDecorator(request){
                @Override
                public Flux<DataBuffer> getBody() {
                    return bodyFlux;
                }
            };
        }

        if( request.getCookies().getFirst("dataServiceUserAccount")!=null){
            account = request.getCookies().getFirst("dataServiceUserAccount").getName();
        }
        TOthGatewayRequestLog getwayLog = new TOthGatewayRequestLog();
        getwayLog.setClientIp(ip);
        getwayLog.setUserId(account);
        getwayLog.setRequestTime(new Date());
        getwayLog.setRequestUrl(url);
        getwayLog.setRequestMethod(methodName);
        getwayLog.setRequestBody(requestBody);
        this.kafkaTemplate.send(GatewayConstant.KAFKA_TOPIC_LOG, JSONUtil.toJsonStr(getwayLog) );
        Integer rule = rules.computeIfAbsent(ip, this::loadRule);

        if (rule == GatewayConstant.ALLOW) {
            return chain.filter(exchange);
        }

        if (rule == GatewayConstant.DENY) {
            response.setStatusCode(HttpStatus.FORBIDDEN);
            return response.setComplete();
        }
        if (rule == GatewayConstant.LIMITED) {
            long timeDiff = System.currentTimeMillis() - timesmillis.computeIfAbsent(ip, k -> System.currentTimeMillis());
            int currCount = counters.get(ip) == null ? 0 : counters.get(ip).intValue();
            log.info(ip + "\t频率范围访问次数=" + currCount + "\t距离上一次访问时间差=" + timeDiff);
            String renewalRate = properties.getRenewalRate();
            int limit = properties.getLimit();
            long millis = Duration.parse("PT" + renewalRate).toMillis();
            //将当前时间与上一次访问时间相差，如果大于刷新频率时间则清空次数
            if (timeDiff > millis) {
                log.info("清空ip访问次数=" + counters.get(ip));
                counters.remove(ip);
            }
            if (counters.computeIfAbsent(ip, k -> new AtomicInteger(0)).incrementAndGet() > limit) {
                response.setStatusCode(HttpStatus.TOO_MANY_REQUESTS);
                return response.setComplete();
            }
        }
        timesmillis.compute(ip, (k, v) -> System.currentTimeMillis());
        return chain.filter(exchange);
    }

    private Integer loadRule(String ip) {
        Integer rule = null;
        List<String> blacklist = properties.getBlacklist();
        List<String> whitelist = properties.getWhitelist();
        Integer limit = properties.getLimit();
        if (whitelist != null && whitelist.contains(ip)) {
            rule = GatewayConstant.ALLOW;
            // 白名单放行
            return rule;
        }

        if (whitelist != null && blacklist.contains(ip)) {
            // 黑名单拦截
            rule = GatewayConstant.DENY;
            return rule;
        }

        //限流
        if (limit != null && limit > 0) {
            rule = GatewayConstant.LIMITED;
        }
        return rule;
    }

    /**
     * 从Flux<DataBuffer>中获取字符串的方法
     * @return 请求体
     */
    private String resolveBodyFromRequest(ServerHttpRequest serverHttpRequest) {
        //获取请求体
        Flux<DataBuffer> body = serverHttpRequest.getBody();
        AtomicReference<String> bodyRef = new AtomicReference<>();
        body.subscribe(buffer -> {
            CharBuffer charBuffer = StandardCharsets.UTF_8.decode(buffer.asByteBuffer());
            DataBufferUtils.release(buffer);
            bodyRef.set(charBuffer.toString());
        });
        //获取request body
        return bodyRef.get();
    }
    /**
     * 字符串转DataBuffer
     * @param value
     * @return
     */
    private DataBuffer stringBuffer(String value) {
        byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
        NettyDataBufferFactory nettyDataBufferFactory = new NettyDataBufferFactory(ByteBufAllocator.DEFAULT);
        DataBuffer buffer = nettyDataBufferFactory.allocateBuffer(bytes.length);
        buffer.write(bytes);
        return buffer;
    }
    @Override
    public int getOrder() {
        return -1;
    }


}
